"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : security-audit
Case Name   : explain相关select操作应被审计到
Create At   : 2023/06/08
Owner       : opentestcase033
Description :
    1.开启select操作审计
    2.创建测试用户和表，赋权
    3.开启表的行访问控制开关，创建表行访问控制策略
    4.测试用户创建打印函数
    5.执行explain语句
    6.查询审计记录
    7.清理环境
Expect      :
    1.设置成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.能正常审计到explain相关select操作
    7.清理环境
History     :
"""

import os
import unittest

from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class Audit(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.cons = Constant()
        self.user = 'u_audit_0009'
        self.table = 't_audit_0009'
        self.func = 'f_audit_0009'

    def test_security(self):
        text = '-----step1:开启select操作审计;expect:成功-----'
        self.log.info(text)
        res = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                        'audit_dml_state_select=1')
        self.assertTrue(res, f"执行失败{text}")

        text = '-----step2:创建用户和表，赋权;expect:成功-----'
        self.log.info(text)
        sql = f'''drop user if exists {self.user} cascade;
        create user {self.user} with password '{macro.COMMON_PASSWD}';
        create table {self.table}(id int, role varchar(100), 
        body text, title text, last_mod_date date);
        insert into {self.table} values(1, 'u_audit_0009', 'China,
         officially the People''s Republic of China(PRC), located in Asia, 
         is the world''s most populous state.', 'China', '2010-1-1'),
         (2, 'xiaoming', 'America is a rock band, formed in England in 
         1970 by multi-instrumentalists Dewey Bunnell, Dan Peek, 
         and Gerry Beckley.', 'America', '2010-1-1'),
         (3, 'xiaoli','England is a country that is part of the 
         United Kingdom. It shares land borders with Scotland to the north 
         and Wales to the west.', 'England','2010-1-1');
        grant select on {self.table} to {self.user};'''
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_ROLE_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.TABLE_CREATE_SUCCESS, msg, f"执行失败:{text}")
        self.assertIn(self.cons.INSERT_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.GRANT_SUCCESS_MSG, msg, f"执行失败:{text}")

        text = '-----step3:开启表的行访问控制开关，创建表行访问控制策略;expect:成功-----'
        self.log.info(text)
        sql = f"alter table {self.table} enable row level security;" \
              f"create row level security policy all_data_rls " \
              f"on {self.table} using(role = current_user);"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(self.cons.ALTER_TABLE_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.CREATE_ROW_LEVEL_SECURITY_POLICY_SUCCESS_MSG,
                      msg, f"执行失败:{text}")

        text = '-----step4:测试用户创建打印函数;expect:成功-----'
        self.log.info(text)
        sql = f'''create or replace function {self.func}(id int,name text)
        returns text language plpgsql IMMUTABLE as \$\$
        begin
        raise notice 'id :% , name : %',id,name;
        return 1;
        end;
        \$\$; '''
        msg = self.pri_sh.execut_db_sql(
            sql, sql_type=f'-U {self.user} -W{macro.COMMON_PASSWD}')
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_FUNCTION_SUCCESS_MSG, msg,
                      f"执行失败:{text}")

        text = '-----step5:执行explain语句;expect:成功-----'
        self.log.info(text)
        explain_sql = f"explain (analyze) select * from {self.table} where " \
                      f"to_tsvector({self.func}(1,title) || ' ' || body) " \
                      f"@@ to_tsquery('china & asia');"
        msg = self.pri_sh.execut_db_sql(
            explain_sql, sql_type=f'-U {self.user} -W{macro.COMMON_PASSWD}')
        self.log.info(msg)
        self.assertIn(f'Seq Scan on {self.table}', msg, f"执行失败:{text}")
        self.assertIn(f'Notice: This query is influenced by row level '
                      f'security feature', msg, f"执行失败:{text}")

        text = '-----step6:查询审计记录;expect:成功-----'
        self.log.info(text)
        sql = f"select pg_sleep(2);" \
              f"select username,type,object_name,detail_info from " \
              f"pg_query_audit(sysdate - interval '20 seconds', sysdate) " \
              f"where username ='{self.user}' and object_name='{self.table}';"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(explain_sql, msg, f"执行失败:{text}")

    def tearDown(self):
        text = '-----step7:清理环境;expect:成功-----'
        self.log.info(text)
        msg = self.pri_sh.execut_db_sql(
            f'drop user if exists {self.user} cascade;'
            f'drop table if exists {self.table} cascade;')
        self.log.info(msg)
        res = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                         'audit_dml_state_select=0')
        self.log.info(res)
        self.assertTrue(res, f"执行失败:{text}")
        self.assertIn(self.cons.TABLE_DROP_SUCCESS, msg, f"执行失败:{text}")
        self.assertIn(self.cons.DROP_ROLE_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
